package com.leezai.mqtt.publish;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

//每次使用都手动new一个客户端，防止并发冲突
public class MqttPublisherClient {

    private static final Logger log = LoggerFactory.getLogger(MqttPublisherClient.class);

    private IMqttAsyncClient mqttClient;

    private static final String clientIdPrefix = "publisher_";

    /**
     * 连接 MQTT Broker
     *
     * @param brokerUrl MQTT Broker地址
     * @param username  用户名
     * @param password  密码
     */
    private void connect(String brokerUrl, String username, String password) throws MqttException, InterruptedException {
        try {
            CountDownLatch connectLatch = new CountDownLatch(1);

            MemoryPersistence persistence = new MemoryPersistence();
            UUID uuid = UUID.randomUUID();
            String clientId = clientIdPrefix + uuid.toString();
            mqttClient = new MqttAsyncClient(brokerUrl, clientId, persistence);

            MqttConnectOptions options = new MqttConnectOptions();
            options.setUserName(username);
            options.setPassword(password.toCharArray());
            options.setAutomaticReconnect(true); // 自动重连

            mqttClient.connect(options, null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    connectLatch.countDown(); // 连接成功，释放等待
                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    throw new RuntimeException("发送失败 MQTT连接失败", exception);
                }
            });

            boolean connected = connectLatch.await(5, TimeUnit.SECONDS);
            if (!connected) {
                throw new RuntimeException("发送失败 MQTT连接超时");
            }
        } catch (MqttException | InterruptedException | RuntimeException e) {
            log.error("MQTT 发送失败: " + e.getMessage());
            throw e;
        }
    }


    /**
     * 发布消息到指定主题
     *
     * @param topic   主题名称
     * @param payload 消息内容
     * @param qos     QoS等级（0, 1, 2）
     */
    private void publish(String topic, byte[] payload, int qos) {
        if (mqttClient != null && mqttClient.isConnected()) {
            try {
                mqttClient.publish(topic, payload, qos, false);
            } catch (MqttException e) {
                log.error("mqtt消息发布失败：" + e.getMessage());
            }
        } else {
            log.error("mqtt消息发布失败：客户端未连接，无法发布消息");
        }
    }

    /**
     * 断开连接
     */
    private void disconnect() {
        if (mqttClient != null && mqttClient.isConnected()) {
            try {
                mqttClient.disconnect();
            } catch (MqttException e) {
                log.error("mqtt断开异常", e);
            }
        }
    }

    /**
     * 发送消息的便捷方法
     *
     * @param brokerUrl MQTT Broker地址
     * @param username  用户名
     * @param password  密码
     * @param topic     主题名称
     * @param payload   消息内容
     * @param qos       QoS等级
     */
    public void send(String brokerUrl, String username, String password,
                     String topic, byte[] payload, int qos) throws MqttException, InterruptedException {
        connect(brokerUrl, username, password);
        publish(topic, payload, qos);
        disconnect();
    }


}
